package pipeline

import (
	"sort"
	"io"
	"encoding/binary"
	"math/rand"
)

const (
	ChunkSizeUnlimited = -1
)

// <-chan int 这里的 `<-`表示只出不进（数据源）
func ArraySource(a ...int) (<-chan int) {
	out := make(chan int)
	go func() {
		for _, v := range a {
			out <- v
		}
		close(out)
	}()
	return out
}

func InMemSort(in <-chan int) <-chan int {
	out := make(chan int, 1024)
	go func() {
		a := make([]int, 0)
		for v := range in {
			a = append(a, v)
		}
		sort.Ints(a)
		for _, v := range a {
			out <- v
		}
		close(out)
	}()
	return out
}

func Merge(in1, in2 <-chan int) chan int {
	out := make(chan int, 1024)
	go func() {
		v1, ok1 := <-in1
		v2, ok2 := <-in2
		for ok1 || ok2 {
			switch {
			// 仅in1有数据
			case !ok2:
				out <- v1
				v1, ok1 = <-in1
				// 仅in2有数据
			case !ok1:
				out <- v2
				v2, ok2 = <-in2
				// 都有数据，且v1<v2
			case v1 < v2:
				out <- v1
				v1, ok1 = <-in1
				// 最后一种情况
			default:
				out <- v2
				v2, ok2 = <-in2
			}
		}
		close(out)
	}()
	return out
}

func MergeN(inputs ...<-chan int) <-chan int {
	l := len(inputs)
	if l == 0 {
		panic("inputs len is zero!")
	}
	if l == 1 {
		return inputs[0]
	}
	m := l / 2
	return Merge(
		MergeN(inputs[:m]...),
		MergeN(inputs[m:]...))
}

func ReaderSource(r io.Reader, chunkSize int) <-chan int {
	out := make(chan int, 1024)
	go func() {
		buffer := make([]byte, 8)
		nread := 0
		for {
			n, err := r.Read(buffer)
			nread += n
			if n < 8 || err != nil ||
				// 限制读取的数量
				(chunkSize != ChunkSizeUnlimited && nread > chunkSize ) {
				break
			}
			v := int(binary.BigEndian.Uint64(buffer))
			out <- v
		}
		close(out)
	}()
	return out
}

func WriterSink(w io.Writer, in <-chan int) {
	for v := range in {
		buffer := make([]byte, 8)
		binary.BigEndian.PutUint64(buffer, uint64(v))
		w.Write(buffer)
	}
}

func RandomSource(count int) <-chan int {

	out := make(chan int)
	go func() {
		for i := 0; i < count; i++ {
			out <- rand.Int()
		}
		close(out)
	}()
	return out

}
